<?php
/**
* producer.php
*
* Create on 2016-12-7
* Create by TT
*/
namespace haibao\cms\kafka;

class Producer extends Base{
	const PAGE_MONITOR_TOPICNAME = 'topic_name';
	
	public function __construct($topicName){
		$this->setTopicName($topicName);
	}
	
	public function setTopicName($topicName){
		self::$topicName = $topicName;
	}
	
	public function getPartitionId(){
		$partitionId = -1;
		$partitions = self::getProduce()->getAvailablePartitions(self::$topicName);
		if ($partitions){
			$partitionId = $partitions[array_rand($partitions, 1)];
		}
		
		return $partitionId;
	}
	
	public function setMessage($message){
		$partitionId = $this->getPartitionId();
		if (self::$topicName && $message && $partitionId != -1){
			self::getProduce()->setMessages(
				self::$topicName,
				$partitionId,
				$message		
			);
		}
	}
	
	public function send(){
		return self::getProduce()->send();
	}
}